package com.spider.moudle.writer;

import com.alibaba.fastjson.JSON;
import com.spider.moudle.api.realize.AbstractWriter;
import com.spider.moudle.entity.Task;
import com.spider.moudle.entity.TaskType;
import com.spider.moudle.factory.KafkaProducerFactory;
import com.spider.moudle.util.DateUtil;
import com.spider.moudle.util.ErrorLogUtil;
import com.spider.moudle.util.PropertiesUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.File;
import java.util.Date;
import java.util.Map;

/**
 * Created by Linbo Hu on 2018/1/15.
 * kafka输出数据
 */
public class KafkaWriter extends AbstractWriter {

    private KafkaProducer<byte[], byte[]> kafkaProducer = KafkaProducerFactory.build();


    @Override
    public boolean write(String path, String fileName, String content) throws Exception {
        throw new Exception("废弃方法，禁止使用");
    }

    @Override
    public boolean write(String path, String fileName, String content, Map<String, File> files) throws Exception {
        throw new Exception("废弃方法，禁止使用");
    }

    @Override
    public boolean write(String path, String fileName, String content, TaskType taskType) throws Exception {
        final boolean[] flag = {true};
        try {
            String speech_id = path + "_" + fileName;
            String data = "{\"content\": "+ content + "," + "\"update_time\":\"" + DateUtil.date_to_str(new Date(), DateUtil.yyyy_MM_dd_HH_mm_ss) + "\"}";
            String topic = null;
            if(TaskType.SPEECH == taskType) {
                topic = PropertiesUtil.getVal("kafka.speech.data.topic");
            } else if(TaskType.USER == taskType) {
                topic = PropertiesUtil.getVal("kafka.user.data.topic");
            } else {
                throw new Exception("task type must be speech or home page");
            }
            if(null != topic) {
                ProducerRecord<byte[],byte[]> kafkaRecord = new ProducerRecord<>(topic, speech_id.getBytes("UTF-8"), data.getBytes("UTF-8"));
                kafkaProducer.send(kafkaRecord, (metadata, e) -> {
                    if(null != e) {
                        flag[0] = false;
                        System.out.println("{method: 'kafka', type: '" + taskType + "', id:'" + speech_id + "', status:'error', nowTime:'" + DateUtil.getNowTimeString(DateUtil.yyyy_MM_dd_HH_mm_ss) + "'}");
                        log.error(e.getMessage(), e);
                        e.printStackTrace();
                    }else {
                        System.out.println("{method: 'kafka', type: '" + taskType + "', id:'" + speech_id + "', status:'complete', nowTime:'" + DateUtil.getNowTimeString(DateUtil.yyyy_MM_dd_HH_mm_ss) + "'}");
                    }
                });
            }
        } catch (Exception e) {
            ErrorLogUtil.error(e.getMessage(), e);
            flag[0] = false;
        }
        return flag[0];
    }

    @Override
    public void write(String taskName, Task task) throws Exception {
        atomicLong.incrementAndGet();
        if(atomicLong.intValue()>MAX_FLOOR){
            atomicLong.set(0);
        }
        String taskTopic = null;
        if(TaskType.SPEECH == task.getTaskType()) {
            taskTopic = PropertiesUtil.getVal("kafka.speech.task.topic");
        } else if(TaskType.USER == task.getTaskType()) {
            taskTopic = PropertiesUtil.getVal("kafka.user.task.topic");
        } else {
            throw new Exception("任务类型不合法");
        }
        String value = JSON.toJSONString(task);
        ProducerRecord<byte[],byte[]> record = new ProducerRecord<>(taskTopic,atomicLong.intValue() % (Integer.parseInt(PropertiesUtil.getVal("kafka.partition.size"))), taskName.getBytes("UTF-8"), value.getBytes("UTF-8"));
        kafkaProducer.send(record, (metadata, e) -> {
            if (e != null) {
                System.out.println("{name: '" + taskName + "', type: '" + task.getTaskType() + "', status:'error',  nowTime:'" + DateUtil.getNowTimeString(DateUtil.yyyy_MM_dd_HH_mm_ss) + "'}");
                log.error(e.getMessage(), e);
                e.printStackTrace();
            } else {
                System.out.println("{name: '" + taskName + "', type: '" + task.getTaskType() + "', status:'success',  nowTime:'" + DateUtil.getNowTimeString(DateUtil.yyyy_MM_dd_HH_mm_ss) + "'}");
            }
        });
    }
}
